package com.bj58.ecdata.monitor.monitor;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.annotation.PostConstruct;

import org.apache.commons.exec.CommandLine;
import org.apache.commons.exec.DefaultExecutor;
import org.apache.commons.exec.ExecuteException;
import org.apache.commons.exec.ExecuteWatchdog;
import org.apache.commons.exec.PumpStreamHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import com.bj58.ecdata.monitor.service.SMS;

/**
 * spark监控，检查job是否挂掉
 * @author zhaoxiang
 */
@Component
public class YarnAppMonitor {
	
	static Logger logger = LoggerFactory.getLogger(YarnAppMonitor.class);


	private static final Map<String, Long> app_start_time = new HashMap<String, Long>();

	private static final Set<String> check_app = new HashSet<String>();
	static {
		check_app.add("CpcClick");
		check_app.add("CpcDisplay");
		check_app.add("TrackM");
		check_app.add("TrackApp");
		check_app.add("TrackPc");
		check_app.add("ImcAdd");
		check_app.add("ResumeAdd");
	}

	@PostConstruct
	public void init() {
		new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						Thread.sleep(1000 * 60 * 1); //1分钟
						process();
					} catch (Exception e) {
						logger.error("YarnAppMonitor process Error: ",e);
					}
				}
			}
		}).start();
	}
	
	//执行命令，获取命令输出结果
	private String cmd(String line) throws ExecuteException, IOException {
		ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
		CommandLine commandline = CommandLine.parse(line);
		DefaultExecutor exec = new DefaultExecutor();
		exec.setExitValues(null);
		ExecuteWatchdog watchdog = new ExecuteWatchdog(30000);
		exec.setWatchdog(watchdog);
		PumpStreamHandler streamHandler = new PumpStreamHandler(outputStream);
		exec.setStreamHandler(streamHandler);
		exec.execute(commandline);// 执行
		String out = outputStream.toString("utf8");
		return out;
	}

	private void process() throws ExecuteException, IOException {
		List<YarnAppInfo> list = yarn_app_infos();
		HashMap<String, Long> status_map = new HashMap<String, Long>();
		
		for (YarnAppInfo one : list) {
			String appId = one.getApplicationId();
			YarnAppStatusInfo status = yarn_app_status_info(appId);
			String app_name = status.getApplicationName();
			Long start_time = Long.parseLong(status.getStartTime());
			status_map.put(app_name, start_time);
		}

		logger.info("yarn_app_infos ======= : " + status_map.toString());
		
		for (String one : check_app) {
			if (status_map.keySet().contains(one)) {
				Long time = status_map.get(one); // 现在的时间
				if (app_start_time.keySet().contains(one)) {
					Long pre_time = app_start_time.get(one);
					if (pre_time.longValue() != time.longValue()) {
						// 说明项目重启过
						app_start_time.put(one, time);
						logger.warn(one + " 实时项目启动了!");
						SMS.defaultSend(one + " 实时项目启动了!");
					}
				} else {
					app_start_time.put(one, time);
				}
			} else {
				logger.warn(one + " 实时项目启停了!");
				SMS.defaultSend(one + " 实时项目停了!");
			}
		}
	}
	
	/**
	 * yarn application -list  信息
	 * @throws ExecuteException
	 * @throws IOException
	 */
	private List<YarnAppInfo> yarn_app_infos() throws ExecuteException,
			IOException {
		List<YarnAppInfo> res = new ArrayList<YarnAppInfo>();
		String command = "yarn application -list";
		String response = cmd(command);
//		logger.error("yarn application -list: \n{}",response);
		String[] array = response.split("\n");
		String rule = "(.+)\t(.+)\t(.+)\t(.+)\t(.+)\t(.+)\t(.+)\t(.+)\t(.+)";
		Pattern pattern = Pattern.compile(rule);
//		logger.error("foreach lines :");
		for (String one : array) {
			logger.error(one);
			Matcher matcher = pattern.matcher(one);
			if (matcher.find()) {
				String applicationId = matcher.group(1).trim();
				String applicationName = matcher.group(2).trim();
				String applicationType = matcher.group(3).trim();
				String user = matcher.group(4).trim();
				String queue = matcher.group(5).trim();
				String state = matcher.group(6).trim();
				String finalState = matcher.group(7).trim();
				String progress = matcher.group(8).trim();
				String trackingURL = matcher.group(9).trim();
				if (!applicationId.equals("Application-Id")) {
					YarnAppInfo yai = new YarnAppInfo(applicationId,
							applicationName, applicationType, user, queue,
							state, finalState, progress, trackingURL);
					res.add(yai);
				}
			}
		}
		return res;
	}

	/**
	 * yarn application -status appId 信息
	 * @throws ExecuteException
	 * @throws IOException
	 */
	private YarnAppStatusInfo yarn_app_status_info(String appId)
			throws ExecuteException, IOException {
		String command = "yarn application -status " + appId;
		String response = cmd(command);
		String[] array = response.split("\n");
		String rule = "(Application-Id|Application-Name|Start-Time) : (.+)";
		Pattern pattern = Pattern.compile(rule);
		YarnAppStatusInfo res = new YarnAppStatusInfo();
		for (String one : array) {
			Matcher matcher = pattern.matcher(one);
			if (matcher.find()) {
				String key = matcher.group(1).trim();
				String value = matcher.group(2).trim();
				if ("Application-Id".equals(key))
					res.setApplicationId(value);
				else if ("Application-Name".equals(key))
					res.setApplicationName(value);
				else if ("Start-Time".equals(key))
					res.setStartTime(value);
			}
		}
		return res;
	}


	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}

class YarnAppStatusInfo {
	private String applicationId;
	private String applicationName;
	private String startTime;

	public YarnAppStatusInfo() {
		super();
	}

	public String getApplicationId() {
		return applicationId;
	}

	public void setApplicationId(String applicationId) {
		this.applicationId = applicationId;
	}

	public String getStartTime() {
		return startTime;
	}

	public void setStartTime(String startTime) {
		this.startTime = startTime;
	}

	public String getApplicationName() {
		return applicationName;
	}

	public void setApplicationName(String applicationName) {
		this.applicationName = applicationName;
	}

	@Override
	public String toString() {
		return "YarnAppStatusInfo ["
				+ (applicationId != null ? "applicationId=" + applicationId
						+ ", " : "")
				+ (applicationName != null ? "applicationName="
						+ applicationName + ", " : "")
				+ (startTime != null ? "startTime=" + startTime : "") + "]";
	}
}

class YarnAppInfo {
	private String applicationId;
	private String applicationName;
	private String applicationType;
	private String user;
	private String queue;
	private String state;
	private String finalState;
	private String progress;
	private String trackingURL;

	public YarnAppInfo() {
		super();
	}

	public YarnAppInfo(String... infos) {
		this.applicationId = infos[0];
		this.applicationName = infos[1];
		this.applicationType = infos[2];
		this.user = infos[3];
		this.queue = infos[4];
		this.state = infos[5];
		this.finalState = infos[6];
		this.progress = infos[7];
		this.trackingURL = infos[8];
	}

	public String getApplicationId() {
		return applicationId;
	}

	public void setApplicationId(String applicationId) {
		this.applicationId = applicationId;
	}

	public String getApplicationName() {
		return applicationName;
	}

	public void setApplicationName(String applicationName) {
		this.applicationName = applicationName;
	}

	public String getApplicationType() {
		return applicationType;
	}

	public void setApplicationType(String applicationType) {
		this.applicationType = applicationType;
	}

	public String getUser() {
		return user;
	}

	public void setUser(String user) {
		this.user = user;
	}

	public String getQueue() {
		return queue;
	}

	public void setQueue(String queue) {
		this.queue = queue;
	}

	public String getState() {
		return state;
	}

	public void setState(String state) {
		this.state = state;
	}

	public String getFinalState() {
		return finalState;
	}

	public void setFinalState(String finalState) {
		this.finalState = finalState;
	}

	public String getProgress() {
		return progress;
	}

	public void setProgress(String progress) {
		this.progress = progress;
	}

	public String getTrackingURL() {
		return trackingURL;
	}

	public void setTrackingURL(String trackingURL) {
		this.trackingURL = trackingURL;
	}

	@Override
	public String toString() {
		return "YarnAppInfo ["
				+ (applicationId != null ? "applicationId=" + applicationId
						+ ", " : "")
				+ (applicationName != null ? "applicationName="
						+ applicationName + ", " : "")
				+ (applicationType != null ? "applicationType="
						+ applicationType + ", " : "")
				+ (user != null ? "user=" + user + ", " : "")
				+ (queue != null ? "queue=" + queue + ", " : "")
				+ (state != null ? "state=" + state + ", " : "")
				+ (finalState != null ? "finalState=" + finalState + ", " : "")
				+ (progress != null ? "progress=" + progress + ", " : "")
				+ (trackingURL != null ? "trackingURL=" + trackingURL : "")
				+ "]";
	}

}
